package org.example.exchange.direct;

import com.rabbitmq.client.*;
import com.sun.org.apache.bcel.internal.generic.ACONST_NULL;
import sun.awt.windows.ThemeReader;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 一个连接可以有多个信道，
 * 多个信道可以对应一个队列
 * 一个队列可以有多个消费者
 */
public class MulitThreadConsumer {

    private static class ConsumerWorker implements Runnable{
        final Connection  connection;
        final String queueName;

        public ConsumerWorker(Connection connection, String queueName) {
            this.connection = connection;
            this.queueName = queueName;
        }

        @Override
        public void run() {
            try {
                final Channel channel = connection.createChannel();
                //定义交换器
                channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,"direct");
                String factQueueName = queueName;
                final String consumerName;
                // 声明一个随机队列
                if(null == factQueueName){
                    factQueueName = channel.queueDeclare().getQueue();
                    consumerName = Thread.currentThread().getName()+"_all";

                }else{
                    // 绑定队列
                    channel.queueDeclare(factQueueName,false,false,false,null);
                    consumerName = Thread.currentThread().getName();
                }
                // 所有日志的严重级别
                String[] severities = {"error","info","warning"};
                for(String severity:severities){
                    // 关注所有级别的日志，多重绑定
                    channel.queueBind(factQueueName,DirectProducer.EXCHANGE_NAME,severity);
                }
                System.out.println("[*] Wating for messages");
                // 创建队列的消费者
                final Consumer consumerA = new DefaultConsumer(channel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                       String message = new String(body,"UTF-8");
                       System.out.println(consumerName+" Receved\t"+envelope.getRoutingKey()+":"+message+"'");
                    }
                };
                channel.basicConsume(factQueueName,true,consumerA);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException,InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");

        // 打开连接和创建频道
        Connection connection = factory.newConnection();
        // 一个连接多个信道，每个队列一个消费者
        for (int i=0;i<2;i++){
            Thread worker = new Thread(new ConsumerWorker(connection,null));
            worker.start();
        }

        /**
         * 3个线程，对应3个信道，3个信道共享一个队列，
         * 线程之间共享队列，一个队列多个消费者，
         */
//        String  queueName = "focusAll";
//        for (int i = 0; i < 3; i++) {
//            Thread worker = new Thread(new ConsumerWorker(connection,queueName));
//            worker.start();
//        }


    }
}
